package com.bml.architect.base;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * 拉模式,就是消费者需要去broker上主动拉取消息
 */
public class PullConsumer {
    private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();

    public static void main(String[] args) {
//这种模式现在已经不推荐使用了，rocketmq还提供了一些新的方式，DefaultLitePullConsumer使用，也可以自己管理消息的offset
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumerPushGrpoup");

        try {
            consumer.setNamesrvAddr(MqConstants.getNameServerAddr());
            consumer.start();
            Set<MessageQueue> topicTest = consumer.fetchSubscribeMessageQueues("TopicTest");
            for (MessageQueue messageQueue : topicTest) {
                NO_MESSAGE:
                for (; ; ) {
                    //自己管理offset，每次默认取32条消息
                    PullResult pullResult = consumer.pullBlockIfNotFound(messageQueue, "*", getMessageQueueOffset(messageQueue), 32);
                    //取完过后，维护offset，就是下次取消息的开始位置，方便下次取
                    putMessageQueueOffset(messageQueue, pullResult.getNextBeginOffset());
                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            //找到消息,处理完成过后又死循环开始从新去broker上拉取消息
                            for (MessageExt messageExt : pullResult.getMsgFoundList()) {
                                System.out.println(new String(messageExt.getBody()));
                            }
                            break;
                        case NO_NEW_MSG:
                            System.out.println("没有新的消息了，退出当前的循环");
                            break NO_MESSAGE;
                        case NO_MATCHED_MSG:
                            break;
                        case OFFSET_ILLEGAL:
                            break;
                    }

                }
            }
            consumer.shutdown();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        }
    }

    private static long getMessageQueueOffset(MessageQueue mq) {
        Long offset = OFFSE_TABLE.get(mq);
        if (offset != null)
            return offset;

        return 0;
    }

    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
        OFFSE_TABLE.put(mq, offset);
    }
}
